/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.zjw.shortlink.project.mq.consumer;

import com.zjw.shortlink.project.common.constant.RedisKeyConstant;
import com.zjw.shortlink.project.common.convention.exception.ServiceException;
import com.zjw.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.zjw.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import com.zjw.shortlink.project.service.ShortLinkService;
import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;

import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.locks.LockSupport;

/**
 * 延迟记录短链接统计组件
 * 公众号：马丁玩编程，回复：加群，添加马哥微信（备注：link）获取项目资料
 */
@Component
@Slf4j
@RequiredArgsConstructor
@Deprecated
public class DelayShortLinkStatsConsumer{
    private final RedissonClient redissonClient;
    private final ShortLinkService shortLinkService;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;

    @PostConstruct
    @SneakyThrows
    public void onMessage() {
        Executors.newSingleThreadExecutor((Runnable r) -> {
            Thread thread = new Thread(r);
            thread.setName("delay_short-link_stats_consumer");
            thread.setDaemon(true);
            return thread;
        }).execute(() -> {
            RBlockingDeque<ShortLinkStatsRecordDTO> blockingDeque = redissonClient.getBlockingDeque(RedisKeyConstant.DELAY_QUEUE_STATS_KEY);
            RDelayedQueue<ShortLinkStatsRecordDTO> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
            while(true) {
                ShortLinkStatsRecordDTO record = delayedQueue.poll();
                if(record != null) {
                    if(messageQueueIdempotentHandler.isMessageProcessed(record.getKeys())) {
                        if(messageQueueIdempotentHandler.isAccomplish(record.getKeys())) {
                            return;
                        }
                        throw new ServiceException("消息未完成流程，需要消息队列重试");
                    }
                    try {
                        shortLinkService.shortLinkStats(record.getFullShortUrl(), record);
                        continue;
                    } catch (Exception ex) {
                        // 处理异常时，删除消息已被处理的标识
                        messageQueueIdempotentHandler.delMessageProcessed(record.getKeys());
                        log.error("延迟记录短链接监控消费异常", ex);
                    }
                    messageQueueIdempotentHandler.setAccomplish(record.getKeys());
                }
                LockSupport.parkUntil(500);
            }
        });
    }
}
